package com.felk.kafka.java;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class E1_KafkaProducer {
    private final static String topic = "kafkabase_test";

    public static void main(String args[]) {
        String messageStr = "这是第%s条数据";
        new InnerProducer(topic).repeat_send(messageStr, 1000);
    }

    public static class InnerProducer {
        private KafkaProducer kafkaProducer;

        private String topic;

        public InnerProducer(String topic) {
            Properties props = new Properties();
            props.put("bootstrap.servers", "192.168.56.101:9092");
            props.put("acks", "all");
            props.put("retries", 0);
            props.put("batch.size", 16384);
            props.put("key.serializer", StringSerializer.class.getName());
            props.put("value.serializer", StringSerializer.class.getName());
            this.kafkaProducer = new KafkaProducer<String, String>(props);
            this.topic = topic;
        }


        public void repeat_send(String msg, int num) {
            for (int i = 0; i < num; i++) {
                String _msg = String.format(msg, i);
                send(_msg);
                System.out.println(_msg);
            }
        }

        public void send(String msg) {
            kafkaProducer.send(new ProducerRecord(topic, msg, msg));
        }
    }


}